package com.googlecode.objectify.hbase;

import com.google.appengine.api.datastore.Entity;
import fi.aalto.hacid.HAcidDelete;
import fi.aalto.hacid.HAcidPut;
import fi.aalto.hacid.HAcidTxn;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import lombok.RequiredArgsConstructor;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author Secondary User
 */
@RequiredArgsConstructor
public class HBaseIndexer implements Runnable
{

	private final HBaseAsyncDatastoreService service;
	BlockingQueue<Entity> toIndex = new LinkedBlockingQueue<Entity>();
	private static final String TO_INDEX_PREFIX = "TI~@";
	private static final byte[] KEY = Bytes.toBytes("K");

	void index(Iterable<Entity> entities) throws Exception
	{
		HAcidTxn txn = new HAcidTxn(service.client);
		for (Entity e : entities)
		{
			toIndex.add(e);
			String keyString = e.getKey().toKeyString();
			HAcidPut put = new HAcidPut(service.index(), Bytes.toBytes(TO_INDEX_PREFIX + keyString));
			put.add(HBaseAsyncDatastoreService.FAMILY, KEY, Bytes.toBytes(keyString));
			for (Map.Entry<String, Object> propertyEntry : e.getIndexedProperties().entrySet())
			{
				put.add(HBaseAsyncDatastoreService.FAMILY, Bytes.toBytes(propertyEntry.getKey()), HBaseTranslator.toBytes(propertyEntry.getValue()));
			}
			txn.put(put);
		}
		txn.commit();
	}

	@Override
	public void run()
	{
		while (true)
		{
			List<Entity> entities = new ArrayList<Entity>();
			toIndex.drainTo(entities); //This blocks until there's something to index.
			HAcidTxn txn = new HAcidTxn(service.client);
			for (Entity e : entities)
			{
				Map<String, Object> properties = e.getIndexedProperties();
				for (Map.Entry<String, Object> entry : properties.entrySet())
				{
					try
					{
						byte[] data = Bytes.add(Bytes.toBytes(entry.getKey()), HBaseTranslator.toBytes(entry.getValue()));
						HAcidPut put = new HAcidPut(service.index(), data);
						put.add(HBaseAsyncDatastoreService.FAMILY, KEY, Bytes.toBytes(e.getKey().toKeyString()));
						txn.put(put);
					}
					catch (Exception ex)
					{
						ex.printStackTrace();
					}
				}
				try
				{
					HAcidDelete delete = new HAcidDelete(service.index(), Bytes.toBytes(TO_INDEX_PREFIX + e.getKey().toKeyString()));
					txn.delete(delete);
				}
				catch (IOException ex)
				{
					ex.printStackTrace();
				}
			}
			try
			{
				txn.commit();
			}
			catch (Exception ex)
			{
				ex.printStackTrace();
			}
		}
	}

	public void deleteFromIndexes(Entity e)
	{
		HAcidTxn txn = new HAcidTxn(service.client);
		Map<String, Object> properties = e.getProperties();
		for (Map.Entry<String, Object> entry : properties.entrySet())
		{
			try
			{
				byte[] data = Bytes.add(Bytes.toBytes(entry.getKey()), HBaseTranslator.toBytes(entry.getValue()));
				HAcidDelete delete = new HAcidDelete(service.index(), data);
				txn.delete(delete);
			}
			catch (Exception ex)
			{
				ex.printStackTrace();
			}
		}
		try
		{
			HAcidDelete delete = new HAcidDelete(service.index(), Bytes.toBytes(TO_INDEX_PREFIX + e.getKey().toKeyString()));
			txn.delete(delete);
		}
		catch (IOException ex)
		{
			ex.printStackTrace();
		}
		try
		{
			txn.commit();
		}
		catch (Exception ex)
		{
			ex.printStackTrace();
		}
	}
}
